package com.zpl.common.support.invoke.impl;

import com.alibaba.fastjson.JSON;
import com.github.houbb.heaven.util.lang.ObjectUtil;
import com.zpl.common.resp.MqCommonRespCode;
import com.zpl.common.resp.MqException;
import com.zpl.common.rpc.RpcMessageDto;
import com.zpl.common.support.invoke.IInvokeService;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author zhupanlin
 * @version 1.0
 * @description: TODO
 * @date 2024/9/8 12:40
 */
@Slf4j
public class InvokeService implements IInvokeService {

    /**
     * 请求序列号 map
     * 1. 如果后期要添加超时检测，可以添加对应的超时时间
     * 
     * key: seqId 唯一标识一个请i去
     * value: 存入该请求最长的有效时间。用于定时删除和超时判断
     */
    private final ConcurrentHashMap<String, Long> requestMap;

    /**
     * 响应结果
     */
    private final ConcurrentHashMap<String, RpcMessageDto> responseMap;
    
    public InvokeService() {
        requestMap = new ConcurrentHashMap<>();
        responseMap = new ConcurrentHashMap<>();

        final Runnable timeoutCheckThread = new TimeoutCheckThread(requestMap, responseMap);
        // 创建定时任务 每隔60秒去扫描请求是否超时
        Executors.newScheduledThreadPool(1)
                .scheduleAtFixedRate(timeoutCheckThread, 60, 60, TimeUnit.SECONDS);
    }

    /**
     * 接收请求
     * @param seqId 序列号
     * @param timeoutMills 超时时间
     * @return
     */
    @Override
    public IInvokeService addRequest(String seqId, long timeoutMills) {
        log.debug("[Invoke] start add request for seqId for seqId: {}, timeoutMills: {}", seqId, timeoutMills);

        final long expireTime = System.currentTimeMillis();
        requestMap.putIfAbsent(seqId, expireTime);
        
        return this;
    }

    /**
     * 接收响应
     * @param seqId 唯一标识
     * @param rpcResponse 响应结果
     * @return
     */
    @Override
    public IInvokeService addResponse(String seqId, RpcMessageDto rpcResponse) {
        // 1.判断是否有效
        Long expireTime = this.requestMap.get(seqId);
        // 如果为空，可能是这个结果已经超时了，被定时任务移除之后，响应结果才过来。直接忽略
        if (ObjectUtil.isNull(expireTime)) {
            return this;
        }
        
        // 2.判断是否超时
        if (System.currentTimeMillis() > expireTime) {
            log.debug("[Invoke] seqId:{}, 直接返回超时结果。", seqId);
            rpcResponse = RpcMessageDto.timeout();
        }
        
        // 这里放入之前，可以添加判断
        // 如果 seqId 必须在处理请求集合中，才允许放入。否则直接忽略丢弃。
        // 通知所有等待方
        responseMap.putIfAbsent(seqId, rpcResponse);
        log.debug("[Invoke] 获取结果信息，seqId: {}，rpcResponse: {}", seqId, JSON.toJSON(rpcResponse));
        log.debug("[Invoke] seqId:{} 信息已经放入，通知所有等待方", seqId);
        
        // 移除对应的 requestMap
        requestMap.remove(seqId);
        log.debug("[Invoke] seqId:{} remove from request map", seqId);
        
        synchronized (this) {
            this.notifyAll();
            log.debug("[Invoke] {} notifyAll()", seqId);
        }


        return this;
    }

    /**
     * 得到响应
     * @param seqId 序列号
     * @return
     */
    @Override
    public RpcMessageDto getResponse(String seqId) {
        
        try {
            RpcMessageDto rpcResponse = this.responseMap.get(seqId);
            if (ObjectUtil.isNotNull(rpcResponse)) {
                log.debug("[Invoke] seq {} 对应结果已经获取: {}", seqId, rpcResponse);
                return rpcResponse;
            }
            
            // 进入等待
            while (rpcResponse == null) {
                log.debug("[Invoke] {} wait has notified!", seqId);

                // 同步等待锁
                synchronized (this) {
                    this.wait();
                }

                log.debug("[Invoke] {} wait has notified!", seqId);

                rpcResponse = this.responseMap.get(seqId);
                log.debug("[Invoke] seq {} 对应结果已经获取: {}", seqId, rpcResponse);
            }

            return rpcResponse;
            
        }catch (Exception e) {
            log.error("获取响应异常", e);
            throw new MqException(MqCommonRespCode.RPC_GET_RESP_FAILED);
        }
        
    }

    @Override
    public boolean remainsRequest() {
        return this.requestMap.size() > 0;
    }
}
